package com.study.mqtt.config;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttSecurityException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.util.StringUtils;

@Configuration
@EnableConfigurationProperties({ MqttProperties.class })
@ConditionalOnProperty(prefix = "mqtt", name = { "enabled" }, matchIfMissing = true)
public class MqttConfig {

	private static final Logger log = LoggerFactory.getLogger(MqttConfig.class);
	protected static final Map<String, MqttMessageConsumer> CONSUMERS = new HashMap<>();

	private final StandardEnvironment environment;

	public MqttConfig(StandardEnvironment environment) {
		this.environment = environment;
	}

	private String resolveClientId() throws UnknownHostException {
//		Integer serverPort = this.environment.getProperty("server.port", Integer.class, 8080);
		String applicationName = this.environment.getProperty("spring.application.name");
		if (!StringUtils.hasText(applicationName)) {
			applicationName = UUID.randomUUID().toString();
		}
		String clientId = applicationName + "@" + applicationName + ":" + InetAddress.getLocalHost().getHostName();
		return clientId;
	}

	@Bean
	public MqttClient mqttClient(MqttProperties properties, MqttConnectOptions options)
			throws UnknownHostException, MqttException {
		MqttClient client = new MqttClient(properties.getBroker(), resolveClientId(), new MemoryPersistence());

		client.setCallback(new MqttCallback() {

			// 连接丢失
			@Override
			public void connectionLost(Throwable cause) {
				log.error("Mqtt broker disconnected.", cause);
				log.debug("Auto reconnect to the border");
				try {
					client.connect(options);
					log.debug("reconnection successfully");
				} catch (MqttException e) {
					log.error("connection error", (Throwable) e);
				}
			}

			// 收到消息
			@Override
			public void messageArrived(String topic, MqttMessage message) throws Exception {
				MqttMessageConsumer mqttMessageConsumer = getMqttMessageConsumer(topic);
				if (null != mqttMessageConsumer) {
					if (log.isDebugEnabled()) {
						log.debug("Message arrived for topic :[{}], qos: [{}], retained :[{}], duplicate :[{}]",
								new Object[] { topic, Integer.valueOf(message.getQos()),
										Boolean.valueOf(message.isRetained()),
										Boolean.valueOf(message.isDuplicate()) });
						log.debug("<< :[{}]", new String(message.getPayload()));
					}
					MDC.put("requestId", UUID.randomUUID().toString());
					try {
						mqttMessageConsumer.consume(topic, message);
					} finally {
						MDC.remove("requestId");
					}
				} else {
					log.warn("No consumer found for topic :[{}]", topic);
				}
			}

			// 消息传递成功
			@Override
			public void deliveryComplete(IMqttDeliveryToken token) {
				log.trace("deliveryComplete complete :{}", Boolean.valueOf(token.isComplete()));
			}

			private MqttMessageConsumer getMqttMessageConsumer(String topic) {
				MqttMessageConsumer mqttMessageConsumer = MqttConfig.CONSUMERS.get(topic);
				if (null != mqttMessageConsumer)
					return mqttMessageConsumer;
				for (Map.Entry<String, MqttMessageConsumer> entry : MqttConfig.CONSUMERS.entrySet()) {
					if (MqttTopic.isMatched(entry.getKey(), topic))
						return entry.getValue();
				}
				return null;
			}
		});
		return client;
	}

	@Bean
	public MqttConnectOptions mqttConnectOptions(MqttProperties properties) throws MqttException {
		MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
		mqttConnectOptions.setUserName(properties.getUsername());
		mqttConnectOptions.setPassword(properties.getPassword().toCharArray());
		mqttConnectOptions.setCleanSession(properties.isCleanSession());
//		mqttConnectOptions.setKeepAliveInterval(10);
//		mqttConnectOptions.setAutomaticReconnect(true);
		return mqttConnectOptions;
	}

}
